package cn.itcast.up.model

import org.apache.commons.lang3.StringUtils
import org.apache.spark.sql.{DataFrame, SparkSession}

/**
  * 测试标签合并
  */
object TestMergeTag {
      var str: String = null
  def main(args: Array[String]): Unit = {

    //创建2个数据源,新数据,老数据
    val spark: SparkSession = SparkSession.builder()
      .appName("TestMergeTag")
      .master("local[*]")
      .getOrCreate()
    //隐式转换
    import spark.implicits._
    import org.apache.spark.sql.functions._

    //加载数据源
    val newDF: DataFrame = List(
      ("100", "485"),
      ("101", "486"),
      ("102", "485"),
      ("103", "487"),
      ("104", "488")
    ).toDF("userid", "tagIds")
    val oldDF: DataFrame = List(
      ("99", "234,555"),
      ("101", "235,223,486"),
      ("102", "345"),
      ("105", "487"),
      ("104", "111,222")
    ).toDF("userid", "tagIds")

    //对新老数据进行数据的合并
    val joinResult: DataFrame = newDF.join(oldDF,newDF.col("userid") === oldDF.col("userid"),"full")

//    joinResult.show()
//+------+------+------+-----------+
    //|userid|tagIds|userid|     tagIds|
    //+------+------+------+-----------+
    //|   101|   486|   101|235,223,486|
    //|  null|  null|    99|    234,555|
    //|   100|   485|  null|       null|
    //|   104|   488|   104|    111,222|
    //|   102|   485|   102|        345|
    //|   103|   487|  null|       null|
    //|  null|  null|   105|        487|
    //+------+------+------+-----------+

    //定义一个函数,接收新老标签数据,进行数据整合
    val mergeTag = udf((newTag: String, oldTag: String) => {
      var tagIds = ""
      //判断新老数据是否为空 isBlank ""
      val string = "" //不是null,是空
      val str1 = "                   " //是空
      if (StringUtils.isBlank(newTag)){
            tagIds = oldTag
      }
      if (StringUtils.isBlank(oldTag)){
        tagIds = newTag
      }
      //如果新老数据都不为空.那么就开始整合
      if (StringUtils.isNotBlank(newTag) && StringUtils.isNotBlank(oldTag)){
        val tmpTag: String = oldTag + "," + newTag
        //将数据使用,进行切割,之后转换为set去重,最后再转换为字符串
        tagIds = tmpTag.split(",").toSet.mkString(",")
      }
      //返回最终结果
      tagIds
    })


    val result: DataFrame = joinResult.select(
      //如果新数据用户ID不为空,就使用新数据的ID
      when(newDF.col("userid").isNotNull, newDF.col("userid"))
        .when(newDF.col("userid").isNull, oldDF.col("userid"))
        .as("userid"),
      mergeTag(newDF.col("tagIds"), oldDF.col("tagIds")).as("tagIds")
    )
    result.show()




  }
}
